package com.chengzhi.disruptor.multi;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 构建多消费者多生产者
 */
public class OrderTest {
    public static void main(String[] args) throws InterruptedException {
        RingBuffer<Order> ringBuffer = RingBuffer.create(ProducerType.MULTI, () -> new Order(), 8, new YieldingWaitStrategy());
        //ringbuffer创建屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        //构建多消费着
        OrderConsumer[] consumer = new OrderConsumer[10];
        for (int i = 0; i < 10; i++) {
            consumer[i] = new OrderConsumer("C" + i);
        }
        //构建多消费着池
        WorkerPool<Order> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, new ExceptionHandler<Order>() {

            /**
             * 消费的时候异常
             * @param ex
             * @param sequence
             * @param event
             */
            @Override
            public void handleEventException(Throwable ex, long sequence, Order event) {

            }

            /**
             * 刚启动的时候
             * @param ex
             */
            @Override
            public void handleOnStartException(Throwable ex) {

            }

            /**
             *  消费结束的时候
             * @param ex
             */
            @Override
            public void handleOnShutdownException(Throwable ex) {

            }
        }, consumer);
        //设置多消费者sequence序号  用于统计消费者进度
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
        //启动
        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));
        AtomicInteger atomicInteger = new AtomicInteger(1);
        for (int i = 0; i < 10; i++) {
            OrderProducer orderProducer = new OrderProducer(ringBuffer);
            new Thread(() -> {
                for (int j = 0; j < 10; j++) {
                    orderProducer.send(atomicInteger.getAndIncrement());
                }
            }).start();
        }
        Thread.sleep(5000);
        System.out.println(consumer[0].getAtomicInteger());
    }
}
